package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSQLApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeOrderDetail().start(10014,4, Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv, String groupId) {

        // 1.读取ODS数据

        readOdsDb(tEnv,groupId);
        // 2. 过滤订单详情流数据
        order_detailStream(tEnv);

        // 3. 过滤订单数据
        order_infoStream(tEnv);

        // 4. 过滤活动流
        order_detail_activityStream(tEnv);

        // 5. 过滤优惠券流
        order_detail_coupon(tEnv);

        // 6. 关联
        Table table = order_join(tEnv);


        // 7. 写入Kafka

        //创建 kafka 主题
        kafkacreate(tEnv);

//        tEnv.sqlQuery("select * from table2024").execute().print();

        //插入kafka
        table.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL).execute();


    }

    //创建要存入 的kafka结构
    private static void kafkacreate(StreamTableEnvironment tEnv) {
        TableResult tableResult = tEnv.executeSql("create table " + Constant.TOPIC_DWD_TRADE_ORDER_DETAIL + "\n" +
                "(\n" +
                " id string,\n" +
                " order_id string,\n" +
                " sku_id string,\n" +
                " user_id string,\n" +
                " province_id string,\n" +
                " activity_id string,\n" +
                " activity_rule_id string,\n" +
                " coupon_id string,\n" +
                " sku_name string,\n" +
                " order_price string,\n" +
                " sku_num string,\n" +
                " create_time string,\n" +
                " split_total_amount string,\n" +
                " split_activity_amount string,\n" +
                " split_coupon_amount string,\n" +
                " split_original_amount string,\n" +
                " ts bigint,\n" +
                "PRIMARY KEY (id) NOT ENFORCED \n" +
                ")" + SQLUtil.getUpsertKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
    }

    //关联
    private static Table order_join(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select \n" +
                "od.id,\n" +
                "order_id,\n" +
                "sku_id,\n" +
                "user_id,\n" +
                "province_id,\n" +
                "activity_id,\n" +
                "activity_rule_id,\n" +
                "coupon_id,\n" +
                "sku_name,\n" +
                "order_price,\n" +
                "sku_num,\n" +
                "create_time,\n" +
                "split_total_amount,\n" +
                "split_activity_amount,\n" +
                "split_coupon_amount,\n" +
                "split_original_amount,\n" +
                "ts\n" +
                "from order_detail od\n" +
                "join  order_info ori\n" +
                "on od.order_id=ori.id\n" +
                "left join coupon_info cou\n" +
                "on od.id =cou.id\n" +
                "left join activity_info ac\n" +
                "on od.id=ac.id");

        tEnv.createTemporaryView("table2024",table);
        return table;
    }

    //过滤订单优惠卷
    private static void order_detail_coupon(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select \n" +
                "`data`['order_detail_id'] id,\n" +
                "`data`['coupon_id'] coupon_id\n" +
                "from topic_db\n" +
                "where  `database` ='gmall' and `table`='order_detail_coupon'\n" +
                "and `type`='insert' ");

        tEnv.createTemporaryView("coupon_info",table);
    }

    //过滤 订单活动
    private static void order_detail_activityStream(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select \n" +
                "`data`['order_detail_id'] id,\n" +
                "`data`['activity_id'] activity_id,\n" +
                "`data`['activity_rule_id'] activity_rule_id\n" +
                "from topic_db\n" +
                "where  `database` ='gmall' and `table`='order_detail_activity'\n" +
                "and `type`='insert' ");

        tEnv.createTemporaryView("activity_info",table);
    }

    // 3. 过滤订单数据
    private static void order_infoStream(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['province_id'] province_id," +
                "`data`['original_total_amount'] split_original_amount\n" +
                "from topic_db\n" +
                "where  `database` ='gmall' and `table`='order_info'\n" +
                "and `type`='insert' ");

        tEnv.createTemporaryView("order_info",table);
    }

    // 2. 过滤订单详情流数据
    private static void order_detailStream(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['sku_name'] sku_name,\n" +
                "`data`['order_price'] order_price,\n" +
                "`data`['sku_num'] sku_num,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['split_total_amount'] split_total_amount,\n" +
                "`data`['split_activity_amount'] split_activity_amount,\n" +
                "`data`['split_coupon_amount'] split_coupon_amount,\n" +
                "ts\n" +
                "from topic_db\n" +
                "where  `database` ='gmall' and `table`='order_detail'\n" +
                "and `type`='insert' ");

        tEnv.createTemporaryView("order_detail",table);
    }
}
